﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Text;

using NMemcached.Client.Configuration;
using NMemcached.Client.Text;

namespace NMemcached.Client
{
    /// <summary>
    /// A memcached server connection using the text protocol
    /// </summary>
    public class TextConnection : ServerConnection
    {
        private static readonly string Error = "ERROR";
        private static readonly string ClientError = "CLIENT_ERROR";
        private static readonly string ServerError = "SERVER_ERROR";
        private static readonly string NotStored = "NOT_STORED";
        private static readonly string Deleted = "DELETED";
        private static readonly string Value = "VALUE";
        private static readonly string End = "END";
        private static readonly string NotFound = "NOT_FOUND";
        private static readonly string Statistic = "STAT";
        private static readonly string Exists = "EXISTS";

        /// <summary>
        /// Initializes a new instance of the TextConnection class
        /// </summary>
        /// <param name="remoteEp">The end point of the memcached server</param>
        public TextConnection(EndPoint remoteEp)
            : this(remoteEp, Encoding.UTF8, false)
        {
        }

        /// <summary>
        /// Initializes a new instance of the TextConnection class
        /// </summary>
        /// <param name="remoteEp">The end point of the memcached server</param>
        /// <param name="encoding">The character encoding for string data sent to and read from the server</param>
        public TextConnection(EndPoint remoteEp, Encoding encoding)
            : this(remoteEp, encoding, false)
        {
        }

        /// <summary>
        /// Initializes a new instance of the TextConnection class
        /// </summary>
        /// <param name="remoteEp">The end point of the memcached server</param>
        /// <param name="encoding">The character encoding for string data sent to and read from the server</param>
        /// <param name="useCompression">Determines if data to be sent to the server is compressed</param>
        public TextConnection(EndPoint remoteEp, Encoding encoding, bool useCompression)
            : base(TransportType.Tcp, remoteEp, encoding, useCompression)
        {
        }

        /// <summary>
        /// Adds a key/value pair to a Memcached server if the data does not already exist.
        /// </summary>
        /// <param name="key">The key to which the data is associated</param>
        /// <param name="data">The data to be stored in the cache</param>
        /// <param name="expiration">The timestamp at wich the entry expires</param>
        /// <returns>The response code of the command sent to the server</returns>
        /// <exception cref="System.ArgumentNullException">key is null</exception>
        public override ResponseCode Add(string key, object data, DateTime expiration)
        {
            try
            {
                return this.SendStorageCommand(OpCode.Add, key, data, expiration);
            }
            catch (ArgumentNullException e)
            {
                throw e;
            }
        }

        /// <summary>
        /// Adds data to an existing cache entry after the existing data
        /// </summary>
        /// <param name="key">The key to which the data is associated</param>
        /// <param name="data">The data to be stored in the cache</param>
        /// <returns>The response code of the command sent to the server</returns>
        /// <exception cref="System.ArgumentNullException">key is null</exception>
        public override ResponseCode Append(string key, object data)
        {
            try
            {
                return this.SendStorageCommand(OpCode.Append, key, data, DateTime.MinValue);
            }
            catch (ArgumentNullException e)
            {
                throw e;
            }
        }

        /// <summary>
        /// Stores data for an existing key only if it has not been updated since the last get
        /// </summary>
        /// <param name="key">The key to which the data is associated</param>
        /// <param name="data">The data to be stored in the cache</param>
        /// <param name="expiration">The timestamp at wich the entry expires</param>
        /// <param name="casUnique">The unique value of an an existing entry</param>
        /// <returns>The response code of the command sent to the server</returns>
        /// <exception cref="System.ArgumentNullException">key is null</exception>
        public override ResponseCode CheckAndSet(string key, object data, DateTime expiration, ulong casUnique)
        {
            try
            {
                return this.SendStorageCommand(OpCode.Set, key, data, expiration, casUnique);
            }
            catch (ArgumentNullException e)
            {
                throw e;
            }
        }

        /// <summary>
        /// Closes the connection to the Memcached server by sending the "quit" command
        /// </summary>
        public override void Close()
        {
            this.SendCommand(this.ProtocolEncoding.GetBytes("quit\r\n"));
            base.Close();
        }

        /// <summary>
        /// Decreases a numeric cached value at the specified key by the specified amount
        /// </summary>
        /// <param name="key">The cache entry to increment</param>
        /// <param name="count">The amount by which the value is decremented</param>
        /// <returns>The new value of the data after the decrement operation was carried out</returns>
        /// <exception cref="System.ArgumentNullException">key is null</exception>
        public override ulong Decrement(string key, ulong count)
        {
            try
            {
                return this.SendIncrementDecrement(key, count, false);
            }
            catch (ArgumentNullException e)
            {
                throw e;
            }
        }

        /// <summary>
        /// Removes a cache entry
        /// </summary>
        /// <param name="key">The key of the entry to be deleted</param>
        /// <returns>The response code of the command sent to the server</returns>
        /// <exception cref="System.ArgumentNullException">key is null</exception>
        public override ResponseCode Delete(string key)
        {
            if (String.IsNullOrEmpty(key))
            {
                throw new ArgumentNullException("key");
            }

            byte[] result = this.SendCommand(this.FormatSingleKeyCommand("delete", key));
            ResponseCode success = ResponseCode.KeyNotFound;
            if (result != null)
            {
                if (this.ProtocolEncoding.GetString(result).StartsWith(TextConnection.Deleted))
                {
                    success = ResponseCode.NoError;
                }
            }
            return success;
        }

        /// <summary>
        /// Immediately invalidates all cached items in all servers
        /// </summary>
        /// <remarks>
        /// <para>The "flush_all" command has no response other than OK unless there is an error</para>
        /// </remarks>
        public override void FlushAll()
        {
            this.SendCommand(this.ProtocolEncoding.GetBytes("flush_all\r\n"));
        }

        /// <summary>
        /// Invalidates all cached items in all servers after the specified delay
        /// </summary>
        /// <param name="delay">The amount of time to pause prior to executing the "flush_all" command.</param>
        /// <remarks>
        /// <para>The "flush_all" command has no response other than OK unless there is an error.</para>
        /// </remarks>
        public override void FlushAll(TimeSpan delay)
        {
            byte[] command = this.FormatSingleKeyCommand("flush_all", delay.TotalSeconds.ToString());
            this.SendCommand(command);
        }

        /// <summary>
        /// Reads a single entry from the cache
        /// </summary>
        /// <param name="key">The key of the item to get from the cache</param>
        /// <returns>The data from the cache associated with the specified key</returns>
        /// <exception cref="System.ArgumentNullException">key is null</exception>
        public override object Get(string key)
        {
            try
            {
                IDictionary<string, object> items = this.Get(new string[] { key });
                object found = null;
                if (items != null && items.Count > 0)
                {
                    if (items.ContainsKey(key))
                    {
                        found = items[key];
                    }
                }
                return found;
            }
            catch (ArgumentNullException e)
            {
                throw e;
            }
        }

        /// <summary>
        /// Reads multiple entries from the cache
        /// </summary>
        /// <param name="keys">A list of keys for which the data is read</param>
        /// <returns>A keyed collection of items read from the cache</returns>
        /// <exception cref="System.ArgumentNullException">A null key was encountered</exception>
        public override IDictionary<string, object> Get(params string[] keys)
        {
            IDictionary<string, CacheDetail> details = this.GetWithDetail(keys);
            IDictionary<string, object> items = null;
            if (details != null && details.Count > 0)
            {
                items = new Dictionary<string, object>();
                foreach (KeyValuePair<string, CacheDetail> pair in details)
                {
                    if (pair.Value == null)
                    {
                        items.Add(pair.Key, null);
                    }
                    else
                    {
                        items.Add(pair.Key, pair.Value.Value);
                    }
                }
            }
            return items;
        }

        /// <summary>
        /// Reads a single entry from the cache and returns the data with detail information
        /// </summary>
        /// <param name="key">The key of the item to get from the cache</param>
        /// <returns>The data from the cache associated with the specified key with detail information</returns>
        /// <exception cref="System.ArgumentNullException">A null key was encountered</exception>
        public override CacheDetail GetWithDetail(string key)
        {
            try
            {
                IDictionary<string, CacheDetail> details = this.GetWithDetail(new string[] { key });
                CacheDetail found = null;
                if (details != null && details.Count > 0)
                {
                    if (details.ContainsKey(key))
                    {
                        found = details[key];
                    }
                }
                return found;
            }
            catch (ArgumentNullException e)
            {
                throw e;
            }
        }

        /// <summary>
        /// Reads multiple entries from the cache with detail information
        /// </summary>
        /// <param name="keys">A list of keys for which the data is read</param>
        /// <returns>A keyed collection of items read from the cache with detail information</returns>
        /// <exception cref="System.ArgumentNullException">A null key was encountered</exception>
        public override IDictionary<string, CacheDetail> GetWithDetail(params string[] keys)
        {
            TextStream resultStream = null;
            IDictionary<string, CacheDetail> items = null;

            //Build the get command

            string command = null;

            try
            {
                command = this.FormatMultiKeyCommand("gets", keys);
            }
            catch (ArgumentNullException e)
            {
                throw e;
            }

            byte[] commandBytes = this.ProtocolEncoding.GetBytes(command);

            try
            {
                //Read the response
                byte[] resultBytes = this.SendCommand(commandBytes);
                //Create a client stream from the response
                resultStream = new TextStream(resultBytes);
                resultStream.Encoding = this.ProtocolEncoding;
                //Convert it to a string for parsing
                string response = this.ProtocolEncoding.GetString(resultBytes);
                //Continue if it's not an error
                this.CheckError(response);
                //Create the dictionary to hold the items
                items = new Dictionary<string, CacheDetail>();
                //Load the keys into the item collection
                foreach (string key in keys)
                {
                    items.Add(key, null);
                }

                //Continue if the response starts with VALUE
                if (response.StartsWith(TextConnection.Value))
                {
                    string currentLine = null;
                    TextGetDetails getDetail = null;
                    byte[] dataBytes = null;
                    IEnumerable<object> deserialized = null;
                    CacheDetail detail = null;
                    //Continue until we see the literal "END"
                    while ((currentLine = resultStream.ReadLine()) != TextConnection.End)
                    {
                        //Only continue if the current line has content
                        if (!String.IsNullOrEmpty(currentLine))
                        {
                            //Get the details of the entry
                            getDetail = this.ParseDetailEntry(currentLine);
                            //Only continue if we have details
                            if (getDetail != null)
                            {
                                dataBytes = new byte[getDetail.Size];
                                resultStream.Read(dataBytes);
                            }
                            //This will only have a value if the details were present
                            //and there is actual data
                            if (dataBytes != null)
                            {
                                //Create the new detail object and set the key
                                detail = new CacheDetail();
                                detail.Key = getDetail.Key;
                                detail.CasUnique = getDetail.CasUnique;
                                //Decompress the value if the compression flag is set
                                if ((getDetail.Flags & StorageFlags.Compressed) == StorageFlags.Compressed)
                                {
                                    dataBytes = this.Compressor.Decompress(dataBytes);
                                }
                                //Deserialize the value if the serialized flag is set
                                if ((getDetail.Flags & StorageFlags.Serialized) == StorageFlags.Serialized)
                                {
                                    deserialized = this.Serializer.Deserialize(dataBytes);
                                    if (deserialized != null)
                                    {
                                        //Set the value to the first value if only one exists
                                        if (deserialized.Count() == 1)
                                        {
                                            detail.Value = deserialized.ElementAt(0);
                                        }
                                        else
                                        {
                                            detail.Value = deserialized;
                                        }
                                    }
                                }
                                else
                                {
                                    //The data has been decompressed, if the flag was set, and
                                    //is not serialized. Try to save it to the detail as a ulong
                                    //if it will parse or the raw string if it will not
                                    string rawValue = this.ProtocolEncoding.GetString(dataBytes);
                                    ulong parsed = 0;
                                    if (ulong.TryParse(rawValue, out parsed))
                                    {
                                        detail.Value = parsed;
                                    }
                                    else
                                    {
                                        detail.Value = rawValue;
                                    }
                                }
                            }
                            //Set the value of the entry with the specified key
                            items[detail.Key] = detail;
                        }
                    }
                }

                return items;
            }
            finally
            {
                if (resultStream != null)
                {
                    resultStream.Close();
                }
            }
        }

        /// <summary>
        /// Increases a numeric cached value at the specified key by the specified amount
        /// </summary>
        /// <param name="key">The cache entry to increment</param>
        /// <param name="count">The amount by which the value is incremented</param>
        /// <returns>The new value of the data after the increment operation was carried out</returns>
        /// <exception cref="System.ArgumentNullException">key is null</exception>
        public override ulong Increment(string key, ulong count)
        {
            try
            {
                return this.SendIncrementDecrement(key, count, true);
            }
            catch (ArgumentNullException e)
            {
                throw e;
            }
        }

        /// <summary>
        /// Adds data to an existing cache entry before the existing data
        /// </summary>
        /// <param name="key">The key to which the data is associated</param>
        /// <param name="data">The data to be stored in the cache</param>
        /// <returns>The response code of the command sent to the server</returns>
        /// <exception cref="System.ArgumentNullException">key is null</exception>
        public override ResponseCode Prepend(string key, object data)
        {
            try
            {
                return this.SendStorageCommand(OpCode.Prepend, key, data, DateTime.MinValue);
            }
            catch (ArgumentNullException e)
            {
                throw e;
            }
        }

        /// <summary>
        /// Replaces the data associated with an existing cache entry
        /// </summary>
        /// <param name="key">The key to which the data is associated</param>
        /// <param name="data">The data to be stored in the cache</param>
        /// <param name="expiration">The timestamp at wich the entry expires</param>
        /// <returns>The response code of the command sent to the server</returns>
        /// <exception cref="System.ArgumentNullException">key is null</exception>
        public override ResponseCode Replace(string key, object data, DateTime expiration)
        {
            try
            {
                return this.SendStorageCommand(OpCode.Replace, key, data, expiration);
            }
            catch (ArgumentNullException e)
            {
                throw e;
            }
        }

        /// <summary>
        /// Adds data to the cache, overwriting any data in an existing entry
        /// </summary>
        /// <param name="key">The key to which the data is associated</param>
        /// <param name="data">The data to be stored in the cache</param>
        /// <param name="expiration">The timestamp at wich the entry expires</param>
        /// <returns>The response code of the command sent to the server</returns>
        /// <exception cref="System.ArgumentNullException">key is null</exception>
        public override ResponseCode Set(string key, object data, DateTime expiration)
        {
            try
            {
                return this.SendStorageCommand(OpCode.Set, key, data, expiration);
            }
            catch (ArgumentNullException e)
            {
                throw e;
            }
        }

        /// <summary>
        /// Reads the cache statistics for the server
        /// </summary>
        /// <returns>The server statistics</returns>
        public override IDictionary<string, string> Stats()
        {
            return this.SendStatsCommand();
        }

        /// <summary>
        /// Gets the version numbers of the server
        /// </summary>
        /// <returns>A string representation of the memcached server version number</returns>
        public override string Version()
        {
            byte[] command = this.ProtocolEncoding.GetBytes("version\r\n");
            byte[] result = this.SendCommand(command);

            string versionString = null;
            if (result != null)
            {
                versionString = this.ProtocolEncoding.GetString(result);
            }
            return versionString;
        }

        //Private Methods
        private void CheckError(string response)
        {
            if (String.IsNullOrEmpty(response))
            {
                throw new ArgumentNullException("response");
            }

            string message = null;
            if (response.StartsWith(TextConnection.Error)
                || response.StartsWith(TextConnection.ClientError)
                || response.StartsWith(TextConnection.ServerError))
            {
                if (response.StartsWith(TextConnection.Error))
                {
                    message = StringTable.Get("error_InvalidTextCommand");
                }
                else
                {
                    int spaceIndex = response.IndexOf(" ");
                    message = response.Substring(spaceIndex + 1);
                }
                throw new MemcachedException(message);
            }
        }

        private TextGetDetails ParseDetailEntry(string entryText)
        {
            if (String.IsNullOrEmpty(entryText))
            {
                throw new ArgumentNullException("entryText");
            }

            TextGetDetails detail = null;
            string[] args = entryText.Split(' ');
            if (args.Length == 5)
            {
                detail = new TextGetDetails();
                //0 = VALUE
                //1 = <key>
                if (!String.IsNullOrEmpty(args[1]))
                {
                    detail.Key = args[1];
                }
                //2 = <flags>
                if (!String.IsNullOrEmpty(args[2]))
                {
                    ushort flags = 0;
                    if (ushort.TryParse(args[2], out flags))
                    {
                        detail.Flags = (StorageFlags)Enum.Parse(typeof(StorageFlags), args[2]);
                    }
                }
                if (!String.IsNullOrEmpty(args[3]))
                {
                    //3 = <size>
                    int size = 0;
                    if (int.TryParse(args[3], out size))
                    {
                        detail.Size = size;
                    }
                }
                if (!String.IsNullOrEmpty(args[4]))
                {
                    //4 = <casUnique>
                    ulong casUnique = 0;
                    if (ulong.TryParse(args[4], out casUnique))
                    {
                        detail.CasUnique = casUnique;
                    }
                }
            }
            else
            {
                throw new FormatException(StringTable.Get("error_BadTextResponse"));
            }
            return detail;
        }

        private ResponseCode SendStorageCommand(OpCode operation, string key, object data, DateTime expiration)
        {
            return this.SendStorageCommand(operation, key, data, expiration, null);
        }

        private ResponseCode SendStorageCommand(OpCode operation, string key, object data, DateTime expiration, ulong? casUnique)
        {
            if (String.IsNullOrEmpty(key))
            {
                throw new ArgumentNullException("key");
            }

            //Create the set command
            byte[] command = this.FormatSetCommand(operation, key, data, expiration, casUnique);

            //Send the command
            ResponseCode result = ResponseCode.NoError;
            byte[] resultBytes = this.SendCommand(command);
            if (resultBytes != null)
            {
                string response = this.ProtocolEncoding.GetString(resultBytes);
                this.CheckError(response);
                if (response.StartsWith(TextConnection.NotStored))
                {
                    result = ResponseCode.ItemNotStored;
                }
                //This can be returned with append, prepend, and cas commands
                if (response.StartsWith(TextConnection.Exists))
                {
                    result = ResponseCode.KeyExists;
                }
            }
            return result;
        }

        private ulong SendIncrementDecrement(string key, ulong count, bool increment)
        {
            if (String.IsNullOrEmpty(key))
            {
                throw new ArgumentNullException("key");
            }

            byte[] command = this.FormatIncrementDecrement(key, count, increment);
            ulong result = ulong.MinValue;
            byte[] resultBytes = this.SendCommand(command);
            if (resultBytes != null)
            {
                string response = this.ProtocolEncoding.GetString(resultBytes);
                this.CheckError(response);
                if (!response.StartsWith(TextConnection.NotFound))
                {
                    result = ulong.Parse(response.Trim());
                }
            }
            return result;
        }

        private IDictionary<string, string> SendStatsCommand()
        {
            TextStream response = null;
            try
            {
                string command = "stats\r\n";
                response = new TextStream(this.SendCommand(this.ProtocolEncoding.GetBytes(command)));
                string currentLine = null;
                string[] args = null;
                IDictionary<string, string> stats = new Dictionary<string, string>();
                while (!response.EndOfStream)
                {
                    currentLine = response.ReadLine();
                    if (!currentLine.StartsWith(TextConnection.End))
                    {
                        if (currentLine.StartsWith(TextConnection.Statistic))
                        {
                            args = currentLine.Split(' ');
                            stats.Add(args[1], args[2]);
                        }
                    }
                }
                return stats;
            }
            finally
            {
                if (response != null)
                {
                    response.Close();
                }
            }
        }

        //Formatting methods for the different command types
        private byte[] FormatSingleKeyCommand(string command, string key)
        {
            return this.ProtocolEncoding.GetBytes(String.Format("{0} {1}\r\n", command, key));
        }

        private string FormatMultiKeyCommand(string command, params string[] keys)
        {
            StringBuilder buffer = new StringBuilder();
            buffer.AppendFormat("{0} ", command);
            foreach (string key in keys)
            {
                buffer.AppendFormat("{0} ", key);
            }
            return String.Concat(buffer.ToString().Trim(), "\r\n");
        }

        private byte[] FormatIncrementDecrement(string key, ulong count, bool increment)
        {
            TextStream outputStream = null;
            try
            {
                outputStream = new TextStream();
                string command = null;
                if (increment)
                {
                    command = "incr";
                }
                else
                {
                    command = "decr";
                }

                outputStream.Write(String.Format("{0} {1} {2}\r\n", command, key, count));

                byte[] output = outputStream.ToArray();
                return output;
            }
            finally
            {
                if (outputStream != null)
                {
                    outputStream.Close();
                }
            }
        }

        private byte[] FormatSetCommand(OpCode command, string key, object data, DateTime expiration)
        {
            return this.FormatSetCommand(command, key, data, expiration, null);
        }

        private byte[] FormatSetCommand(OpCode command, string key, object data, DateTime expiration, ulong? casUnique)
        {
            using (TextStream outputStream = new TextStream())
            {
                //Pick the correct command for the set operation
                if (casUnique == null)
                {
                    switch (command)
                    {
                        case OpCode.Add:
                            outputStream.Write("add ");
                            break;
                        case OpCode.Append:
                            outputStream.Write("append ");
                            break;
                        case OpCode.Prepend:
                            outputStream.Write("prepend ");
                            break;
                        case OpCode.Replace:
                            outputStream.Write("replace ");
                            break;
                        case OpCode.Set:
                            outputStream.Write("set ");
                            break;
                    }
                }
                else
                {
                    outputStream.Write("cas ");
                }
                outputStream.Write(String.Concat(key, " "));

                //Default to no flags
                //These flags are ignored for append and prepend operations
                StorageFlags flags = StorageFlags.None;
                //Only flag for serialization if it's not a primitive or a string
                if (!data.GetType().IsPrimitive)
                {
                    if (!typeof(string).IsAssignableFrom(data.GetType()))
                    {
                        //Set the serialized flag if not a primitive or string
                        flags |= StorageFlags.Serialized;
                    }
                    //Only compress serialized types or strings
                    if (this.UseCompression)
                    {
                        flags |= StorageFlags.Compressed;
                    }
                }

                //Write the flags to the output stream
                outputStream.Write(String.Concat((ushort)flags, " "));

                //Set the expiration if specified
                if (expiration > DateTime.MinValue)
                {
                    outputStream.Write(String.Concat(ServerConnection.GetExpirationTime(expiration), " "));
                }
                else
                {
                    outputStream.Write("0 ");
                }

                //Assemble the cache item as bytes
                byte[] dataBytes = null;
                if (data.GetType().IsPrimitive)
                {
                    dataBytes = this.PayloadEncoding.GetBytes(Convert.ToUInt64(data).ToString());
                }
                else
                {
                    if (typeof(string).IsAssignableFrom(data.GetType()))
                    {
                        dataBytes = this.PayloadEncoding.GetBytes(data.ToString());
                    }
                    else
                    {
                        //Serialize any non primitive
                        dataBytes = this.Serializer.Serialize(data);
                    }
                    //Compress the data prior to sending it
                    if (this.UseCompression)
                    {
                        dataBytes = this.Compressor.Compress(dataBytes);
                    }
                }

                //Write the tail of the command
                outputStream.Write(dataBytes.Length.ToString());
                //Write the casUnique if available
                if (casUnique != null)
                {
                    outputStream.Write(String.Format(" {0}", casUnique));
                }
                //write the data
                outputStream.Write("\r\n");
                outputStream.Write(dataBytes);
                outputStream.Write("\r\n");

                byte[] commandBytes = outputStream.ToArray();
                return commandBytes;
            }
        }
    }
}